package com.niit.Hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class TableAmdinTest {
    //声明全局变量
    private Configuration configuration;
    private Connection connection;
    private Admin admin;

    //创建连接方法
    @Before
    public void beforeTest() throws IOException {
        //通过HBaseConfiguration得到一个真实的配置对象
        configuration = HBaseConfiguration.create();
        //通过HADOOP配置
        connection = ConnectionFactory.createConnection(configuration);
        //通过连接就可以获得一个操作管理员对象
        admin = connection.getAdmin();
    }

    //创建命名空间
    @Test
    public void createNamespace(){
        //1.调用方法创建命名空间
        // 代码相对 shell 更加底层 所以 shell 能够实现的功能 代码一定能实现
        NamespaceDescriptor.Builder builder =
                NamespaceDescriptor.create("MYBD2");

        // 2. 使用 builder 构造出对应的添加完参数的对象 完成创建
        // 创建命名空间出现的问题 都属于本方法自身的问题 不应该抛出
        try {
            admin.createNamespace(builder.build());
        } catch (IOException e) {
            System.out.println("命令空间已经存在");
            e.printStackTrace();
        }

    }

    //判断表是否存在
    @Test
    public void isTableExists(){
        boolean b = false;
        try {
            b = admin.tableExists(TableName.valueOf("MYBD2",
                    "studnet"));
            System.out.println(b);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建表
    // 创建一个名为WATER_BILL的表，包含一个列蔟C1
    @Test
    public void createTable() throws IOException {
        //名称空间名
        String NAME_SPACE = "MYBD2";
        // 表名
        String TABLE_NAME = "studnet";
        // 列蔟名
        String COLUMN_FAMILY = "C1";

        // 1. 判断表是否存在
        if(admin.tableExists(TableName.valueOf(NAME_SPACE,TABLE_NAME))) {
            return;
        }

        // 2. 构建表描述构建器
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(NAME_SPACE,TABLE_NAME));

        // 3. 构建列蔟描述构建器
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(COLUMN_FAMILY));

        // 4. 构建列蔟描述
        ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();//想法.创建()-->物质

        // 5. 构建表描述
        // 添加列蔟
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);//将实质性的列簇放在表的想法
        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();

        // 6. 创建表
        admin.createTable(tableDescriptor);
    }

    //修改表
    @Test
    public void modifyTable() throws IOException {
        //名称空间名
        String NAME_SPACE = "MYBD2";
        // 表名
        String TABLE_NAME = "studnet";
        // 列蔟名
        String COLUMN_FAMILY = "C2 ";
        int VERSION = 2;

        // 1. 判断表是否存在
        if(admin.tableExists(TableName.valueOf(NAME_SPACE,TABLE_NAME))) {
            return;
        }

        TableDescriptor descriptor =
                admin.getDescriptor(TableName.valueOf(NAME_SPACE, TABLE_NAME));
        // 2.1 创建一个表格描述建造者
        // 如果使用填写 tableName 的方法 相当于创建了一个新的表格描述建造者 没有之前的信息
        // 如果想要修改之前的信息 必须调用方法填写一个旧的表格描述
        TableDescriptorBuilder tableDescriptorBuilder =
                TableDescriptorBuilder.newBuilder(descriptor);
        // 2.2 对应建造者进行表格数据的修改
        ColumnFamilyDescriptor columnFamily1 =
                descriptor.getColumnFamily(Bytes.toBytes(COLUMN_FAMILY));
        // 创建列族描述建造者
        // 需要填写旧的列族描述
        ColumnFamilyDescriptorBuilder
                columnFamilyDescriptorBuilder =
                ColumnFamilyDescriptorBuilder.newBuilder(columnFamily1);
        // 修改对应的版本
        columnFamilyDescriptorBuilder.setMaxVersions(VERSION);

        // 此处修改的时候 如果填写的新创建 那么别的参数会初始化
        tableDescriptorBuilder.modifyColumnFamily(columnFamilyDescriptorBuilder.build());
        admin.modifyTable(tableDescriptorBuilder.build());
    }

    //删除表
    @Test
    public void dropTable() throws IOException {
        //名称空间名
        String NAME_SPACE = "MYBD2";
        // 表名
        String TABLE_NAME = "studnet";
        // 表名     通过Java代码，获得一个hbase的表
        TableName tableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);

        // 1. 判断表是否存在
        if(admin.tableExists(tableName)) {
            // 2. 禁用表
            admin.disableTable(tableName);
            // 3. 删除表
            admin.deleteTable(tableName);
        }
    }

    //插入数据
    @Test
    public void addTest() throws IOException {
        //名称空间名
        String NAME_SPACE = "MYBD2";
        // 表名
        String TABLE_NAME = "studnet";

        // 1.使用Hbase连接获取Htable
        TableName waterBillTableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);
        Table waterBillTable = connection.getTable(waterBillTableName);

        // 2.构建ROWKEY、列蔟名、列名
        String rowkey = "1001";
        String cfName = "C1";
        String colName = "NAME";

        // 3.构建Put对象（对应put命令）
        Put put = new Put(Bytes.toBytes(rowkey));

        // 4.添加姓名列
        put.addColumn(Bytes.toBytes(cfName)
                , Bytes.toBytes(colName)
                , Bytes.toBytes("帅哥严"));

        // 5.使用Htable表对象执行put操作
        waterBillTable.put(put);
        // 6. 关闭表
        waterBillTable.close();

    }

    //查询一条数据
    @Test
    public void getDataByRowkey() throws IOException{
        //名称空间
        String NAME_SPACE = "MYBD2";
        //表名
        String TABLE_NAME = "studnet";
        //1.使用hbaseL连接获取Htable
        TableName waterBillTableName = TableName.valueOf(NAME_SPACE,TABLE_NAME);
        Table waterBilltable = connection.getTable(waterBillTableName);
        //2.使用rowkey构建GET对象
        Get get = new Get(Bytes.toBytes("1001"));
        //3.执行get请求
        Result result = waterBilltable.get(get);
        //4. 获取所有的单元格
        List<Cell> cellList = result.listCells();

        //打印rowkey
        System.out.println("rowkey =>"+Bytes.toString(result.getRow()));

        //5.迭代单元格列表
        for (Cell cell : cellList){
            //打印列簇名
            System.out.println(Bytes.toString(cell.getQualifierArray(),cell.getFamilyOffset()));
            System.out.println("=>"+ Bytes.toString(cell.getValueArray(),cell.getFamilyOffset(),cell.getFamilyLength()));
        }


    }

    //关闭资源，释放资源
    @After
    public void afterTest() throws IOException {
        admin.close();
        connection.close();
    }
}
